package ldwu.spark.ml

import org.apache.spark.mllib.clustering.KMeans
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.{SparkConf, SparkContext}
/**
 * Created by liangdong.wu on 2015/11/30.
 */
object KMeansML {
  def main(args: Array[String]) {
    if (args.length < 2) {
      System.err.println("Usage: <file> <file>")
      System.exit(1)
    }

    val conf = new SparkConf()
    val sc = new SparkContext(conf)
    val data = sc.textFile(args(0))
//    val parsedData = data.map(s=>if (s.trim.isEmpty) None else s.split(' ')).filter(a => if (a == None) false else true).collect()
//    val parsedData = data.map(s => s.split(' ').filter(s => if (s.trim.isEmpty) false else true)).filter(a=>if (a.isEmpty) false else true).collect()
    val parsedData = data.map(s => Vectors.dense(s.split(' ').map(_.toDouble))).cache()

    // Cluster the data into two classes using KMeans
    val numClusters = 2
    val numIterations = 20
    val clusters = KMeans.train(parsedData, numClusters, numIterations)

    // Evaluate clustering by computing Within Set Sum of Squared Errors
    val WSSSE = clusters.computeCost(parsedData)
    println("Within Set Sum of Squared Errors = " + WSSSE)

    // Save and load model
    clusters.save(sc, args(1))
//    val sameModel = KMeansModel.load(sc, args(1))
    sc.stop()
  }

}
